package executors;

import il.technion.ewolf.kbr.KeybasedRouting;
import il.technion.ewolf.kbr.Node;
import il.technion.ewolf.kbr.openkad.KadNetModule;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Random;
import java.util.Set;

import quasar.Station;
import quasar.Statistics;
import quasar_framework.Common;
import quasar_framework.KeyToIDMapper;
import tweeter.NetworkEntity;

import com.google.inject.Guice;
import com.google.inject.Injector;

public class SyntheticQuasarExecutorRevised
{
	/**
	 * The function is used in order to generate the stations participating in the network setup.
	 * All the variables / required configuration is fetched from the JointDefenitions module.
	 * 
	 * @return A list of entities (Stations encapsulated) to be used for the scenario.
	 */
	public static ArrayList<Station> createStations(boolean useRouteTable)
	{
		System.out.println("*** Entered createEntities ***");
		ArrayList<Station> stations = new ArrayList<Station>();
		
		try
		{
			// Create the stations
			for (int index = 0 ; index < Common.numStations ; index++)
			{
				Injector injector = Guice.createInjector(new KadNetModule()
					.setProperty("openkad.keyfactory.keysize", Common.keySize.toString())
					.setProperty("openkad.bucket.kbuckets.maxsize", Common.maxBucketsNum.toString())
					.setProperty("openkad.executors.server.nrthreads", Common.maxServerThreadsNum.toString())
					.setProperty("openkad.executors.client.nrthreads", Common.maxClientThreadsNum.toString())
					.setProperty("openkad.executors.ping.nrthreads", Common.maxPingThreadsNum.toString())
					.setProperty("openkad.executors.client.max_pending", Common.maxClientMessageQueueSize.toString())
					.setProperty("openkad.executors.server.max_pending", Common.maxServerMessageQueueSize.toString())
					.setProperty("openkad.executors.ping.max_pending", Common.maxPingMessageQueueSize.toString())
					.setProperty("openkad.seed", ""+(index+Common.basePort))
					.setProperty("openkad.net.udp.port", ""+(index+Common.basePort)));
			        KeybasedRouting kbr = injector.getInstance(KeybasedRouting.class);
			        kbr.create();
					Station station = new Station(index, kbr, useRouteTable);
					stations.add(station);
			}
		}
		catch (Exception e)
		{
			System.out.println("The system has encountered an exception while trying to create the entities. Error is " + e.getLocalizedMessage());
			System.out.println("*** Exited (Error) createEntities ***");
			return null;
		}
		
		System.out.println("*** Exited createEntities ***");
		return stations;
	}
	
	/**
	 * The first version of the random joining method. In this version, it is first decided which stations
	 * join which stations, and afterwards, a list of URIs is created in order to apply the joining to each
	 * station once.
	 * 
	 * @param entities - The list of entities (stations)
	 * @param stationsToJoin - How many stations each station will join
	 */
	public static void randomJoinv1(ArrayList<NetworkEntity> entities, int stationsToJoin)
	{
		System.out.println("*** Entered randomJoin ***");
		int whoToJoin[][] = new int[Common.numStations][Common.numStations];
		for (int i = 0 ; i < Common.numStations ; i++)
			for (int j = 0 ; j < Common.numStations ; j++)
				whoToJoin[i][j] = -1;
		Random generator = new Random();
		for (int index = 0 ; index < Common.numStations ; index++)
		{
			for (int iteration = 0 ; iteration < stationsToJoin ; iteration++)
			{
				int entityToJoin = generator.nextInt(Common.numStations);
				while (index == entityToJoin || -1 != whoToJoin[entityToJoin][index])
				{
					entityToJoin = generator.nextInt(Common.numStations);
				}
				whoToJoin[entityToJoin][index] = 0;
			}
		}
		
		for (int index = 0 ; index < Common.numStations ; index++)
		{
			ArrayList<URI> stationJoiners = new ArrayList<URI>();
			for (int index2 = 0 ; index2 < Common.numStations ; index2++)
			{
				if (0 == whoToJoin[index][index2])
				{
					try
					{
						stationJoiners.add(new URI("openkad.udp://127.0.0.1:"+(Common.basePort + index2)+"/"));
					}
					catch (URISyntaxException e) 
					{
						System.out.println("An error has been encountered while trying to join, joining will be skipped. Error is " + e.getLocalizedMessage());
					}
				}
			}
			if (!stationJoiners.isEmpty())
			{
				((Station)entities.get(index)).kbr.join(stationJoiners);
			}
		}
		
		System.gc();
		
		sleepy(5000);
		
		System.out.println("*** Exited randomJoin ***");
	}
	
	/**
	 * The second version of the random joining method. In this version, the joining process is performed one
	 * station at a time. When a station finds the desired station it wishes to join to, the joining is invoked.
	 * (The number of joins to be performed for each station is according to the stationsToJoin variable)
	 * 
	 * @param entities - The list of entities (stations)
	 * @param stationsToJoin - How many stations each station will join
	 */
	public static void randomJoinv2(ArrayList<Station> stations, int stationsToJoin)
	{
		System.out.println("*** Entered randomJoin ***");
		Random generator = new Random();
		for (int index = 0 ; index < Common.numStations ; index++)
		{
			Set<Integer> alreadyJoined = new HashSet<Integer>();
			for (int iteration = 0 ; iteration < stationsToJoin ; iteration++)
			{
				int stationToJoin = generator.nextInt(Common.numStations);
				while (index == stationToJoin || alreadyJoined.contains(stationToJoin))
				{
					stationToJoin = generator.nextInt(Common.numStations);
				}
				alreadyJoined.add(stationToJoin);
				
				try
				{
					stations.get(stationToJoin).kbr.join(Arrays.asList(new URI("openkad.udp://127.0.0.1:"+(Common.basePort + index)+"/")));
				}
				catch (URISyntaxException e)
				{
					System.out.println("An error has been encountered while trying to join, joining will be skipped. Error is " + e.getLocalizedMessage());
				}
			}
		}
		
		sleepy(5000);
		
		System.out.println("*** Exited randomJoin ***");
	}
	
	/**
	 * The function is used in order to establish a symmetrical P2P network, since it is essential for
	 * Quasar. KadModule doesn't necessary provide symmetry.
	 * @param stations
	 */
	public static void createSymmetricalNetwork(ArrayList<Station> stations)
	{
		System.out.println("*** Entered createSymmetricalNetwork ***");
		for (int index = 0 ; index < Common.numStations ; index++)
		{
			Station currentStation = stations.get(index);
			List<Node> kbrNeighbours = currentStation.kbr.getNeighbours();
			for (Node neighbour : kbrNeighbours)
			{
				try
				{
					currentStation.kbr.sendMessage(neighbour, "HI", null);
				} 
				catch (IOException e)
				{
					System.out.println("Station " + index + " has encountered an error while trying to send a P2P message. Error is " + e.getLocalizedMessage());
				}
			}
		}
		
		sleepy(5000);
		
		System.out.println("*** Exited createSymmetricalNetwork ***");
	}
	
	public static void generateSubscriptions(ArrayList<Station> stations, int subscriptionsForEach)
	{
		System.out.println("*** Entered generateSubscriptions ***");
		Random generator = new Random();
		int subscribersNum[] = new int[Common.numStations];
		for (int stationIndex = 0 ; stationIndex < Common.numStations ; stationIndex++) subscribersNum[stationIndex] = 0;
		
		for (int stationIndex = 0 ; stationIndex < Common.numStations ; stationIndex++)
		{
			HashSet<Integer> stationSubscriptions = new HashSet<Integer>();
			for (int randomIndex = 0 ; randomIndex < subscriptionsForEach ; randomIndex++)
			{
				int subscription = generator.nextInt(Common.numStations);
				while ((stationSubscriptions.contains(subscription)) || 
					   (stationIndex == subscription))
				{
					subscription = generator.nextInt(Common.numStations);
				}
				// Adding the subscription in the station's list
				stationSubscriptions.add(subscription);
				
				// Updating the statistics
				Statistics.addSubcription(subscription);
				subscribersNum[subscription]++;
			}
			stations.get(stationIndex).subscriptions = stationSubscriptions;
			System.out.println("Station " + stationIndex + " subscriptions: " + stationSubscriptions);
		}
		System.out.println("*** Exited generateSubscriptions ***");
	}
	
	public static void generateSubscriptions(ArrayList<Station> stations, int publisherID, int stationsToSubscribe)
	{
		System.out.println("*** Entered generateSubscriptions ***");
		Random generator = new Random();
		
		HashSet<Integer> stationSubscription = new HashSet<Integer>(); 
		stationSubscription.add(publisherID);
		
		ArrayList<Integer> subscribedStations = new ArrayList<Integer>();
		for (int stationIndex = 0 ; stationIndex < stationsToSubscribe ; stationIndex++)
		{
			int stationID = generator.nextInt(Common.numStations);
			while ((stationID == publisherID) || subscribedStations.contains(stationID)) 
			{
				stationID = generator.nextInt(Common.numStations);
			}
			subscribedStations.add(stationID);
			stations.get(stationID).subscriptions = stationSubscription;
			Statistics.addSubcription(publisherID);
		}
		
		System.out.println("Stations subscribed to " + publisherID + " are " + subscribedStations);
		
		System.out.println("*** Exited generateSubscriptions ***");
	}
	
	
	public static void spreadSubscriptions(ArrayList<Station> stations)
	{
		for (int index = 0 ; index < Common.numStations ; index ++)
		{
			stations.get(index).sendAllYourSubscriptions();
		}
			
		System.out.println("*** Exited generateSubscriptions ***");
	}
	
	public static void makeEveryonePublish(ArrayList<Station> stations)
	{
		for (int stationIndex = 0 ; stationIndex < Common.numStations ; stationIndex++)
		{
			System.out.println("[***************] Station " + stationIndex + " publish [***************]");
			Station st = stations.get(stationIndex);
			st.publish(stationIndex, null);
			sleepy(1000);
		}
	}
	
	public static void makeSinglePublish(ArrayList<Station> stations, int publisherID)
	{
		System.out.println("[***************] Station " + publisherID + " publish [***************]");
		Station st = stations.get(publisherID);
		st.publish(publisherID, null);
		sleepy(1000);
	}
	
	/**
	 * The function is used in order establish a well-known network of subscription information. Each station
	 * spams its own subscription requests on the network. In order to avoid a flood, a minimal pause of 1 second
	 * will be used between subscription sends. Additionally to invoking the subscriptions, the function also
	 * randomly generates the subscriptions for each station.
	 * 
	 * @param stations - The entities (Stations) in the system
	 * @param publishers - A list of decided publishers.
	 */
	public static void generateSubscriptions(ArrayList<Station> stations, ArrayList<Integer> publishers)
	{
		System.out.println("*** Entered generateSubscriptions ***");
		Random generator = new Random();
		
		for (int stationIndex = 0 ; stationIndex < Common.numStations ; stationIndex++)
		{
			int numberOfSubscriptions = generator.nextInt(publishers.size() / 2) + 1;
			HashSet<Integer> stationSubscriptions = new HashSet<Integer>();
			for (int randomIndex = 0 ; randomIndex < numberOfSubscriptions ; randomIndex++)
			{
				int publisherIndex = generator.nextInt(publishers.size());
				while ((stationSubscriptions.contains(publishers.get(publisherIndex))) || 
					   (stationIndex == publishers.get(publisherIndex)))
				{
					publisherIndex = generator.nextInt(publishers.size());
				}
				// Adding the subscription in the station's list
				stationSubscriptions.add(publishers.get(publisherIndex));
				
				// Updating the statistics
				Statistics.addSubcription(publishers.get(publisherIndex));
			}
			stations.get(stationIndex).subscriptions = stationSubscriptions;
		}
		
//		for (int index = 0 ; index < Common.numStations ; index ++)
//		{
//			stations.get(index).sendAllYourSubscriptions();
//			sleepy(10);
//		}
		
//		System.gc();
		
//		sleepy(60000 * (Common.numStations / 100));
		
		System.out.println("*** Exited generateSubscriptions ***");
	}
	
	/**
	 * The function is used in order to generate the stations which will have the role of publishers in the
	 * system. The function assumes that the input won't be higher than the number of stations.
	 * 
	 * @param numPublishers - The number of publishers to be assigned.
	 * @return A list of publisher IDs
	 */
	public static ArrayList<Integer> generatePublishers(int numPublishers)
	{
		System.out.println("*** Entered generatePublishers ***");
		Random generator = new Random();
		
		ArrayList<Integer> publishers = new ArrayList<Integer>();
			
		for (int entry = 0 ; entry < numPublishers ; entry++)
		{
			int publisher = generator.nextInt(Common.numStations);
			while (publishers.contains(publisher))
			{
				publisher = generator.nextInt(Common.numStations);
			}
			publishers.add(publisher);
		}
		System.out.println("*** Exited generatePublishers ***");
		return publishers;
	}
	
	/**
	 * The function is used in order to gracefully teardown the system
	 * 
	 * @param entities - The list of entities (Stations).
	 */
	public static void cleanUp(ArrayList<Station> stations)
	{
		System.out.println("*** Entered cleanUp ***");
		for (int index = 0 ; index < Common.numStations ; index++)
		{
			try
			{
				stations.get(index).kbr.shutdown();
			}
			catch (Exception e)
			{
				System.out.println("The system has encountered an error while trying to cleanup station " + index + ". Error is " + e.getLocalizedMessage());
			}
		}
		
		System.gc();
		System.out.println("*** Exited cleanUp ***");
	}
				
	private static ArrayList<Station> commonInit(boolean useRouteTable)
	{
		Statistics.init(Common.numStations);
		
		ArrayList<Station> stations = createStations(useRouteTable);
		
		if (null == stations)
		{
			return null;
		}

		randomJoinv2(stations, 2);
		
		createSymmetricalNetwork(stations);
				
		for (int index = 0 ; index < Common.numStations ; index++)
		{
			for (int level = 0 ; level < Common.radius ; level++)
			{
				stations.get(index).fillNeighbors(level);
			}
		}
		
		return stations;
	}
	
	public static void invokePublicationsThread(ArrayList<Station> stations, ArrayList<Integer> publishers)
	{
		PublicationsThread publicator = new PublicationsThread(publishers, stations);
		System.out.println("Starting publications...");
		publicator.start();
		System.out.println("Going to sleep for like 2.5 minutes...");
		sleepy(150000);
		publicator.halt();
		int sleepTime = 180000 * (Common.numStations / 100);
		System.out.println("[Stopped publications!] Going to sleep for like " + (sleepTime/1000) + " more minutes, to let the messages settle...");
		sleepy(sleepTime);		
	}
	
	public static void commonTeardown(ArrayList<Station> stations)
	{
		cleanUp(stations);
		stations.clear();
	}
	
	public static void printStatistics()
	{
		System.out.println("-------------------------------------");
		System.out.println("Number of stations=" + Common.numStations);
		System.out.println("Statistics for radius=" +Common.radius + " Pulbication TTL=" + Common.publicationTTL);
		System.out.println("Generated publications: " + Statistics.sentPublications);
		System.out.println("Total publication traffic: " + Statistics.publishPacketTraffic);
		System.out.println("Delivered publications: " + Statistics.deliveredPublishPackets);
		System.out.println("Publications expected to be delivered: " + Statistics.expectedPublishPacketsRecievers);
		System.out.println("Delivery ratio: " + (((float) Statistics.deliveredPublishPackets) / Statistics.expectedPublishPacketsRecievers));
		System.out.println("-------------------------------------");
	}
	
	@SuppressWarnings("static-access")
	/**
	 * This function has been created in order to avoid the copy&paste of the try&catch clause.
	 * @param sleepingTime
	 */
	public static void sleepy(int sleepingTime)
	{
		try
		{
			Thread.currentThread().sleep(sleepingTime);
		}
		catch (InterruptedException e)
		{
			System.out.println("Threads sleep has been interrupted.");
		}
	}

	public static void verifyNeighborsSending(ArrayList<Station> stations)
	{
		for (int i = 0 ; i < Common.numStations ; i++)
		{
			Station s = stations.get(i);
			System.out.println("Station " + i + " info");
			for (int l = 0 ; l < Common.radius ; l++)
			{
				System.out.println("Level " + l);
				HashSet<Node> hs = s.radiusNeighbors.get(l);
				for (Node n : hs)
				{
					System.out.println("Neighbor " + s.mapz.get(n.getKey()) + " : Originator " + s.mapz.get(s.originators.get(n)));
				}
			}
		}
	}
	
	public static void verifyNoKeyMix(ArrayList<Station> stations)
	{
		for (int i = 0 ; i < Common.numStations ; i++)
		{
			for (int j = i+1 ; j < Common.numStations ; j++)
			{
				if (stations.get(i).kbr.getLocalNode().getKey().equals(stations.get(j).kbr.getLocalNode().getKey()))
				{
					System.out.println("XXXXXXXXXXX " + i + " SAME " + j + "XXXXXXXXXXX");
				}
			}
		}
	}
	
	/**
		Assign each station with a set of subscriptions, 
		invoke a publication from each station (once) and run without the routing table.				
	 */
	public static void scenario1()
	{
		ArrayList<Station> stations = commonInit(false);

		generateSubscriptions(stations, 10);
		
		makeEveryonePublish(stations);
		
		printStatistics();
		
		commonTeardown(stations);
	}

	/**
		Assign each station with a set of subscriptions, 
		invoke a publication from each station (once) and run with the routing table.	
     */			
	public static void scenario2()
	{
		ArrayList<Station> stations = commonInit(true);
		
		generateSubscriptions(stations, 10);
		
		spreadSubscriptions(stations);
		
		makeEveryonePublish(stations);
		
		printStatistics();
		
		commonTeardown(stations);
	}
	
	/**
		Assign each station with a set of subscriptions, 
		invoke periodic random publications and run without the routing table.

	 */
	public static void scenario3()
	{
		ArrayList<Station> stations = commonInit(false);
		
		ArrayList<Integer> publishers = generatePublishers(Common.numStations / 10);
		
		generateSubscriptions(stations, publishers);
			
		invokePublicationsThread(stations, publishers);
		
		printStatistics();
		
		commonTeardown(stations);
	}
	
	/**
		Assign each station with a set of subscriptions, 
		invoke periodic random publications and run with the routing table.
	 */
	public static void scenario4()
	{
		ArrayList<Station> stations = commonInit(true);
		
		ArrayList<Integer> publishers = generatePublishers(Common.numStations / 10);
		
		generateSubscriptions(stations, publishers);
		
		spreadSubscriptions(stations);
		
		invokePublicationsThread(stations, publishers);
		
		printStatistics();
		
		commonTeardown(stations);
	}

	/**
		Select a set of stations, 
		and assign them a subscription, 
		and make that station publish without the routing table.
	 */
	public static void scenario5()
	{
		ArrayList<Station> stations = commonInit(false);
		
		int randomPublisher = generatePublishers(1).get(0);
		
		generateSubscriptions(stations, randomPublisher, Common.numStations / 10);
		
		makeSinglePublish(stations, randomPublisher);
		
		printStatistics();
		
		commonTeardown(stations);
	}
	
	/**
		Select a set of stations, 
		and assign them a subscription, 
		and make that station publish with the routing table
	 */
	public static void scenario6()
	{
		ArrayList<Station> stations = commonInit(true);
		
		int randomPublisher = generatePublishers(1).get(0);
		
		generateSubscriptions(stations, randomPublisher, Common.numStations / 10);
		
		spreadSubscriptions(stations);
		
		makeSinglePublish(stations, randomPublisher);
		
		printStatistics();
		
		commonTeardown(stations);
	}
	
	public static void main(String[] args)
	{
		
		Common.numStations = 500;
		Common.parallelPublishersNum = 3;
		Common.keySize = 6;
		Common.maxBucketsNum = 6;
		
		Common.publicationTTL = 4;
		Common.radius = 3;
		
		scenario6();

	}
}
